package com.lvmama.rhino.common.utils.kafka

import org.apache.kafka.clients.producer.KafkaProducer

import scala.collection.mutable

/**
  * Created by yuanxiaofeng on 2016/6/23.
  */
object KafkaProducerFactory {

  import scala.collection.JavaConversions._
  private val producers = mutable.Map[Map[String, String], KafkaProducer[String, String]]()

  def getOrCreateProducer(config: Map[String, String]): KafkaProducer[String, String] ={

    val defaultConfig = Map(
      "key.serializer" -> "org.apache.kafka.common.serialization.StringSerializer",
      "value.serializer" -> "org.apache.kafka.common.serialization.StringSerializer"
    )

    val finalConfig = defaultConfig ++ config

    producers.getOrElseUpdate(finalConfig, {
      val producer = new KafkaProducer[String, String](finalConfig)

      sys.addShutdownHook{
        producer.close()
      }

      producer
    })
  }
}
